import { Component, OnInit } from '@angular/core';
import { Observable, Subject, asapScheduler, pipe, of, from, interval, merge, fromEvent, timer, observable, bindCallback } from 'rxjs';
import { map, filter, scan, take, concatAll, concatMap, throttleTime, delayWhen, delay } from 'rxjs/operators';
import { webSocket } from 'rxjs/webSocket';
import { ajax } from 'rxjs/ajax';
import { TestScheduler } from 'rxjs/testing';
// import 'rxjs-compat/operator/let';

@Component({
    selector: 'rxjs-startup',
    templateUrl: './startup.component.html',
    // styleUrls: ['./name.component.scss']
})
export class StartupComponent implements OnInit {
    // RxCommonJs = require('rxjs');
    // ObCommon = require('rxjs/Observable').Observable;
    constructor() { }

    ngOnInit(): void {
        // 普通例子
        // const onSubscribe = observer => {
        //     observer.next(1);
        //     observer.next(2);
        //     observer.next(3);
        // };
        // const source$ = new Observable(onSubscribe);
        // const theObserver = {
        //     next: item => console.log(item)
        // }
        // source$.subscribe(theObserver);
        // of(1, 2, 3).subscribe(console.log);

        // timer例子
        // const onSubscribeTimer = observer => {
        //     let number = 1;
        //     const handle = setInterval(() => {
        //         observer.next(number++);
        //         if (number > 3) {
        //             clearInterval(handle);
        //             observer.complete();
        //         }
        //     }, 1000)
        //     return {
        //         unsubscribe: () => {
        //             clearInterval(handle);
        //         }
        //     }
        // };
        // const timer$ = new Observable(onSubscribeTimer);
        // const theTimerObserver = {
        //     next: item => console.log(item),
        //     complete: () => console.log('observer complete..'),
        //     error: error => console.error(error)
        // }
        // timer$.subscribe(theTimerObserver);

        // interval(1000).pipe(take(3)).subscribe((value) => console.log(value + 1));

        // unsub例子
        // const onSubscribeTimer2 = observer => {
        //     let number = 1;
        //     const handle = setInterval(() => {
        //         console.log('in sub' + number);
        //         observer.next(number++);
        //     }, 1000)
        //     return {
        //         unsubscribe: () => {
        //             clearInterval(handle);
        //         }
        //     }
        // };
        // const timer2$ = new Observable(onSubscribeTimer2);
        // const theTimerObserver2 = {
        //     next: item => console.log(item),
        //     complete: () => console.log('observer complete..'),
        //     error: error => console.error(error)
        // }
        // let subscription = timer2$.subscribe(theTimerObserver2);
        // setTimeout(() => {
        //     subscription.unsubscribe();
        // }, 3500);

        // 热流例子
        // let p = new Producer(0);
        // p.run();
        // const onSubscribeHot = observer => {
        //     setInterval(() => {
        //         observer.next(p.currnetNum);
        //     }, 999)
        // };
        // const hot$ = new Observable(onSubscribeHot);
        // const theHotObserver = {
        //     next: item => console.log('hot:' + item)
        // }
        // setTimeout(() => {
        //     hot$.subscribe(theHotObserver)
        // }, 3500);


        // 冷流例子
        // const onSubscribeCold = observer => {
        //     let p = new Producer(0);
        //     p.run();
        //     setInterval(() => {
        //         observer.next(p.currnetNum);
        //     }, 999)
        // };
        // const cold$ = new Observable(onSubscribeCold);
        // const thColdObserver = {
        //     next: item => console.log('cold:' + item)
        // }
        // setTimeout(() => {
        //     cold$.subscribe(thColdObserver)
        // }, 3500);


        // 自定义操作符
        // const op = map2.bind(of(3, 4, 5));
        // const results$ = op(x => x);
        // results$.subscribe(console.log);

        // const net$ = ajax.getJSON('http://npm.paic.com.cn:7072/v1/fs/WIT/msgDefinition');

        // timer(2000).subscribe(() =>
        //     net$.subscribe((data) => { console.log(data); })
        // );

        // const ho$ = interval(1000).pipe(
        //     take(2),
        //     map(x => interval(1500).pipe(
        //         map(y => `${x}: ${y}`),
        //         take(2))
        //     ),
        //     concatAll()
        // );
        // ho$.subscribe(y => console.log(y));

        const ho3$ = interval(1000).pipe(
            take(2),
            concatMap(x => interval(1500).pipe(
                map(y => `${x}: ${y}`),
                take(2)
            ))
        );
        ho3$.subscribe(y => console.log(y));


        // const srouce$ = timer(1000, 2000);
        // const filter$ = srouce$.pipe(
        //     delay(1000),
        //     throttleTime(2000));
        // filter$.subscribe(x => console.log(x));
        // const cb = (message) => { console.log('bind callback ' + message); };
        // const bindCB$ = bindCallback(cb);
        // bindCB$.call(cb, 'message').subscribe(() => { console.log('sub'); });a
    }
}

export class Producer {
    currnetNum: number;

    constructor(index: number) {
        this.currnetNum = index;
    }

    run() {
        setInterval(() => {
            this.currnetNum++;
        }, 1000);
    }
}

export function map2() {
    return new Observable(observer => {
        const sub = this.subscribe({
            next: value => {
                try {
                    observer.next(value * 2);
                } catch (error) {
                    observer.error(error);
                }
            },
            complete: () => observer.complete(),
            error: error => observer.error(error)
        });
        return {
            unsubscribe: () => {
                sub.unsubscribe();
            }
        }
    })
}
(<any>Observable.prototype).map = map;

export function map3() {
    // this === Observable
    return this.lift(function (source$) {
        const sub = source$.subscribe({
            next: value => {
                try {
                    this.next(value + 1);
                } catch (error) {
                    this.error(error);
                }
            },
            complete: () => this.complete(),
            error: error => this.error(error)
        });
        return {
            unsubscribe: () => {
                sub.unsubscribe();
            }
        }
    });
}